package org.jetlinks.pro.assets;

import org.jetlinks.pro.assets.event.AssetsBindEvent;
import org.jetlinks.pro.assets.event.AssetsUnBindAllEvent;
import org.jetlinks.pro.assets.event.AssetsUnBindEvent;
import org.reactivestreams.Publisher;
import org.springframework.context.ApplicationEventPublisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;

public class CompositeAssetBindManager implements AssetBindManager {

    final List<AssetBindManager> managers = new CopyOnWriteArrayList<>();

    private final ApplicationEventPublisher eventPublisher;

    public CompositeAssetBindManager(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }

    public void addManager(AssetBindManager manager) {
        if (manager == this) {
            return;
        }
        managers.add(manager);
    }

    @Nonnull
    @Override
    public Mono<Integer> unbindAllAssets(@Nonnull String assetType,
                                         @Nonnull Collection<?> assetId) {
        return Flux
            .fromIterable(managers)
            .flatMap(manager -> manager.unbindAllAssets(assetType, assetId))
            .reduce(Math::addExact)
            .flatMap(AssetsUnBindAllEvent
                         .of(assetType, assetId.stream().map(String::valueOf).collect(Collectors.toSet()))
                         .publish(eventPublisher)
                         ::thenReturn);
    }

    @Nonnull
    @Override
    public Mono<Void> bindAssets(@Nonnull Publisher<AssetBindRequest> requestFlux) {
        Flux<AssetBindRequest> cache = Flux.from(requestFlux).cache();
        //event
        Mono<Void> then = cache
            .flatMap(request -> AssetsBindEvent
                .of(request.getTargetType(),
                    request.getTargetId(),
                    request.getAssetType(),
                    request.getAssetIdList(),
                    request.getPermission(),
                    request.getRelation(),
                    request.isAllPermission())
                .publish(eventPublisher))
            .then();

        return Flux
            .fromIterable(managers)
            .flatMap(manager -> manager.bindAssets(cache))
            .then(then);
    }

    @Override
    public Mono<Void> changePermission(@Nonnull String targetType,
                                       @Nonnull Collection<String> targetIdList,
                                       @Nonnull AssetType assetType,
                                       @Nonnull Collection<?> assetIdList,
                                       @Nonnull AssetPermission ... permissions) {
        return Flux
            .fromIterable(managers)
            .flatMap(manager -> manager.changePermission(targetType,targetIdList,assetType,assetIdList,permissions))
            .then();
    }

    @Nonnull
    @Override
    public Mono<Void> unbindAssets(@Nonnull Publisher<AssetUnbindRequest> requestFlux) {
        Flux<AssetUnbindRequest> cache = Flux.from(requestFlux).cache();
        //event
        Mono<Void> then = cache
            .flatMap(request -> AssetsUnBindEvent
                .of(request.getTargetType(), request.getTargetId(), request.getAssetType(), request.getAssetIdList())
                .publish(eventPublisher))
            .then();

        return Flux
            .fromIterable(managers)
            .flatMap(manager -> manager.unbindAssets(cache))
            .then(then);
    }


    @Nonnull
    @Override
    public Flux<AssetBinding> getBindings(@Nonnull String targetType,
                                          @Nonnull Collection<String> targetIdList,
                                          @Nonnull AssetType assetType,
                                          @Nonnull Collection<?> assetIdList) {
        return Flux.fromIterable(managers)
                   .flatMap(manager -> manager.getBindings(targetType, targetIdList, assetType, assetIdList));
    }

    @Nonnull
    @Override
    public Flux<AssetBinding> getBindings(@Nonnull Collection<String> targetTypes,
                                          @Nonnull AssetType assetType,
                                          @Nonnull Collection<?> assetIdList) {
        return Flux.fromIterable(managers)
                   .flatMap(manager -> manager.getBindings(targetTypes, assetType, assetIdList));
    }

    @Nonnull
    @Override
    public Flux<AssetBinding> getBindings(@Nonnull String targetType,
                                          @Nonnull Collection<String> targetIdList,
                                          @Nonnull AssetType assetType) {
        return Flux.fromIterable(managers)
                   .flatMap(manager -> manager.getBindings(targetType, targetIdList, assetType));
    }

    @Nonnull
    @Override
    public Flux<AssetBinding> getBindings(@Nonnull AssetType assetType, @Nonnull Collection<?> assetIdList) {
        return Flux.fromIterable(managers)
                   .flatMap(manager -> manager.getBindings(assetType, assetIdList));
    }
}
